package com.headcaselabs.orchestra;

import java.io.FileInputStream;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.headcaselabs.queue.SQSUtils;
import com.headcaselabs.work.UnitOfWork;
import com.xerox.amazonws.sqs.SQSException;
/**
 * JobDaemon - entry point to the distributed computing system.
 * @author Florian Leibert
 *
 */
public class JobDaemon implements Runnable{
	protected long sleepTime;
	protected Object lock = new Object();
	protected String hostname;
	protected boolean running = true;
	protected SQSUtils sqsUtils;
	protected Configuration conf;
	private long startTime;
	private long stopTime;
	private static JobDaemon singleton = null;
  private final static Log log_ =
    LogFactory.getLog(JobDaemon.class.getName());

  

	
	public static JobDaemon getInstance() {
		return singleton;
	}
	
	
	public static JobDaemon buildNewInstance(SQSUtils utils, Configuration conf) {
		singleton = new JobDaemon(utils,conf);
		return singleton;
	}
	
	private JobDaemon(SQSUtils utils, Configuration conf) {
		this.sqsUtils = utils;
		this.conf = conf;
		this.sleepTime = conf.getSleepOnEmptyQueues();
	}
	
	/*private JobDaemon(FileInputStream fis) {
		XStream xstream = new XStream();
		xstream.alias("configuration", Configuration.class);
		conf = (Configuration)xstream.fromXML(fis);
		sqsUtils = new SQSUtils(conf);
		rd = new ReportDaemon(conf.getReportInterval(),sqsUtils);
		singleton = this;
	}*/
	
	public void run() {
		UnitOfWork j = null, uof = null;
		startTime = System.currentTimeMillis();
		stopTime = startTime + conf.getWorkingTime();
		log_.debug("Starting time:"+startTime);
		log_.debug("Stop time:"+stopTime);
		int i = 0;
		
		//double load;
		boolean executerAvailable = true;
		//RunLinuxCommand rlc = new RunLinuxCommand();
		//boolean cacheEmpty = true;
		try {
			while (running) {
				log_.debug("Time left until termination:"+(stopTime-System.currentTimeMillis())+" ms");
				if (System.currentTimeMillis()>=stopTime) {
					running = false;
					log_.debug("Time limit for JobDaemon exceeded. Stopping...");
					break;
				}
				if(uof==null) {
					log_.debug("Getting next job from the queue");
					uof = sqsUtils.nextJob();
					if (uof!=null) {
						log_.debug("Found a job in the queue. ");
						continue;
					}
					else {
						log_.debug("No job found in the queue. Falling asleep for:"+sleepTime+"ms");
						Thread.sleep(sleepTime);
					}
				} else {
					synchronized(lock) {
						if (executerAvailable = JobUtils.isReady()) {
							log_.debug("Executor ready, submitting job:"+sleepTime+"ms");
							JobUtils.runMonitoredJob(uof);
							uof = null;
							log_.debug("Job Submitted...");
						} else {
							log_.debug("Executor busy. Falling asleep for:"+sleepTime+"ms");
							Thread.sleep(sleepTime);
						}
					}
				}
			}
		} catch (SQSException sqse) {
			log_.error(sqse);
		} catch (InterruptedException ie) {
			log_.error(ie);
		}
		
		/**
		
		while(running) {
			try {
				while(conf.getMaxLoad()<(load=Double.parseDouble(rlc.execute(conf.getLoadScript())))) {
					log_.info("Encountering high load. Going to sleep...");
					Thread.sleep(conf.getSleepOnEmptyQueues());
					log_.info("Awoke, retrying...");
				}
			} catch (IOException ie) {
				log_.error(ie);
			} catch (InterruptedException ire) {
				log_.error(ire);
			}
			log_.debug("Iteration:"+i++);
			if (System.currentTimeMillis()>=stopTime) {
				running = false;
				log_.info("System ran for "+(System.currentTimeMillis()-startTime)+" ms. Stopping Job Daemon...");
				break;
			} else {
				log_.debug("Time left until termination:"+(stopTime-System.currentTimeMillis())+" ms");
			}
			try {
				if (cacheEmpty) {
					j = sqsUtils.nextJob();
					log_.info("Received job:"+j);
				} 
				log_.info("About to enter locked section...");
				synchronized(lock) {
					log_.debug("Checking if executor is ready..."+JobUtils.isReady());
					log_.debug("Currently there are: " + JobUtils.getExecutor().getActiveCount() + " active threads");
					if (executerAvailable = JobUtils.isReady()) {
						if (j!=null) {
							JobUtils.runMonitoredJob(j);
							log_.debug("Job Submitted...");
							cacheEmpty = true;
						}
					} else {
						cacheEmpty = false;
						log_.debug("Cache full. Thread awaiting to be executed...");
						JobUtils.getExecutor().purge();
					}
				}
				if (!cacheEmpty&&!executerAvailable) {
					log_.debug("Cache full. Sleeping.. for "+sleepTime+" ms, time left:"+(stopTime-System.currentTimeMillis()));
					Thread.sleep(sleepTime);
				}
				else if (j==null) {
					log_.debug("No job found. About to fall asleep for:"+sleepTime+"ms, time left:"+(stopTime-System.currentTimeMillis()));
					Thread.sleep(sleepTime);			
				}
			} catch (SQSException sqse) {
				sqse.printStackTrace();
			} catch (InterruptedException ie) {
				ie.printStackTrace();
			}
		}**/
		
		log_.debug("Done running the Job Daemon.");
		
	}
	
	public SQSUtils getSQSUtils() {
		return sqsUtils;
	}
	
	public Configuration getConfiguration() {
		return conf;
	}
	
	public static void main(String args[]) throws Exception {
		System.out.println("Deprecated way of using the Job Daemon.");
		System.exit(1);
		if (args.length!=1) {
			System.out.println("Usage: java JobDaemon configFile");
			System.exit(1);
		}
		FileInputStream fis = new FileInputStream(args[0]);
		/*JobDaemon gd = new JobDaemon(fis);
		Thread t0 = new Thread(gd);
		t0.start();
		*/
	}

}
